package com.wno704.intf.service.impl;

import com.alibaba.druid.pool.DruidDataSourceFactory;
import com.ctg.mq.api.impl.MQProducerImpl;
import com.wno704.intf.config.WnoConfig;
import com.wno704.intf.service.DbSourceService;
import com.wno704.system.entity.DbSource;
import com.wno704.system.util.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * @ClassName DataSourceServiceImpl
 * @Description TODO
 * @Author wno704
 * @Date 2022/10/21 14:24
 * @Version 1.0
 */
@Slf4j
@Service("dbSourceService")
public class DbSourceServiceImpl implements DbSourceService {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    WnoConfig wnoConfig;

    @Override
    public int init() {
        String sql = "SELECT SOURCE_ID, TYPE, JDBCTEMPLATE_NAME, DRIVER_CLASS_NAME, URL, USERNAME, PASSWORD, CONFIG, STATUS FROM " + wnoConfig.getTablePrefix() + "DBSOURCE  WHERE STATUS = '0' ";
        List<DbSource> dbSourceList = this.jdbcTemplate.query(sql, new BeanPropertyRowMapper<DbSource>(DbSource.class));
        for (DbSource ds : dbSourceList) {
            loadDbSourceBean(ds);
        }
        return dbSourceList.size();
    }

    public void loadDbSourceBean(DbSource ds) {
        try {
            SpringContextUtil.destroyBean(ds.getJdbctemplateName());
            updateDbSource(ds);
            if (ds.getStatus().equals(DbSource.runStatus.DEL.getValue())) {
                return;
            }
            if ("KAFKA".equals(ds.getType())) {
                Map<String, Object> props = new HashMap<>();
                props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ds.getUrl());
                props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
                props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
                ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(props);
                SpringContextUtil.registerBean(ds.getJdbctemplateName(), KafkaTemplate.class, producerFactory);
            } else if ("CTGMQ".equals(ds.getType())) {
                Properties properties = new Properties();
                properties.setProperty("producerGroupName", ds.getDriverClassName());
                properties.setProperty("namesrvAddr", ds.getUrl());
                properties.setProperty("authId", ds.getUsername());
                properties.setProperty("authPwd", ds.getPassword());
                if (StringUtils.isNotBlank(ds.getConfig())) {
                    String[] s = ds.getConfig().split("\n");
                    for (String ss : s) {
                        String[] ag = ss.split("\\|");
                        properties.setProperty(ag[0], ag[1]);
                    }
                }
                //IMQProducer producer = CTGMQFactory.createProducer(properties);
                SpringContextUtil.registerBean(ds.getJdbctemplateName(), MQProducerImpl.class, properties);
            } else if ("RabbitMQ".equals(ds.getType())) {
                CachingConnectionFactory cf = new CachingConnectionFactory();

                String[] ipt = ds.getUrl().split(":");
                cf.setHost(ipt[0]);
                cf.setPort(Integer.parseInt(ipt[1]));

                cf.setUsername(ds.getUsername());
                cf.setPassword(ds.getPassword());
                cf.setVirtualHost(ds.getDriverClassName());

                String queueVal = "";
                String exchangeVal = "";

                if (StringUtils.isNotBlank(ds.getConfig())) {
                    String[] s = ds.getConfig().split("\n");
                    for (String ss : s) {
                        String[] ag = ss.split("\\|");
                        if ("queue".equals(ag[0].toLowerCase())) {
                            queueVal = ag[1];
                        }
                        if ("exchange".equals(ag[0].toLowerCase())) {
                            exchangeVal = ag[1];
                        }
                    }
                }

                RabbitAdmin admin = new RabbitAdmin(cf);
                if (StringUtils.isNotBlank(queueVal)) {
                    Queue queue = new Queue(queueVal);
                    admin.declareQueue(queue);
                }
                if (StringUtils.isNotBlank(exchangeVal)) {
                    TopicExchange exchange = new TopicExchange(exchangeVal);
                    admin.declareExchange(exchange);
                }
                SpringContextUtil.registerBean(ds.getJdbctemplateName(), RabbitTemplate.class, cf);
            } else if ("ActiveMQ".equals(ds.getType())) {
                ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(ds.getUrl());
                SpringContextUtil.registerBean(ds.getJdbctemplateName(), RabbitTemplate.class, cf);
            } else if ("RocketMQ".equals(ds.getType())) {
                Map<String, Object> props = new HashMap<>();
                props.put("namesrvAddr", ds.getUrl());
                props.put("producerGroup", ds.getDriverClassName());
                props.put("instanceName", ds.getUsername());
                SpringContextUtil.registerBean(ds.getJdbctemplateName(), DefaultMQProducer.class, props);
            } else if ("ElasticSearch".equals(ds.getType())) {
                String scheme = ds.getDriverClassName();
                String[] hostList = ds.getUrl().split(";");
                RestClientBuilder builder = null;
                if (hostList.length > 2) {
                    builder = RestClient.builder(new HttpHost(hostList[0].split(":")[0], Integer.parseInt(hostList[0].split(":")[1]), scheme),
                            new HttpHost(hostList[1].split(":")[0], Integer.parseInt(hostList[1].split(":")[1]), scheme),
                            new HttpHost(hostList[2].split(":")[0], Integer.parseInt(hostList[2].split(":")[1]), scheme));
                } else if (hostList.length == 2) {
                    builder = RestClient.builder(new HttpHost(hostList[0].split(":")[0], Integer.parseInt(hostList[0].split(":")[1]), scheme),
                            new HttpHost(hostList[1].split(":")[0], Integer.parseInt(hostList[1].split(":")[1]), scheme));
                } else {
                    builder = RestClient.builder(new HttpHost(hostList[0].split(":")[0], Integer.parseInt(hostList[0].split(":")[1]), scheme));
                }
                if (StringUtils.isNotBlank(ds.getUsername())) {
                    final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                    credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(ds.getUsername(), ds.getPassword()));  //es账号密码
                    builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                        public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                            httpClientBuilder.disableAuthCaching();
                            return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                        }
                    }).setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
                        // 该方法接收一个RequestConfig.Builder对象，对该对象进行修改后然后返回。
                        @Override
                        public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                            return requestConfigBuilder.setConnectTimeout(5000 * 1000) // 连接超时（默认为1秒）
                                    .setSocketTimeout(6000 * 1000);// 套接字超时（默认为30秒）//更改客户端的超时限制默认30秒现在改为100*1000分钟
                        }
                    });
                }
                SpringContextUtil.registerBean(ds.getJdbctemplateName(), RestHighLevelClient.class, builder, true);
            } else {
                Map<String, String> map = new HashMap();
                map.put(DruidDataSourceFactory.PROP_URL, ds.getUrl());
                map.put(DruidDataSourceFactory.PROP_DRIVERCLASSNAME, ds.getDriverClassName());
                map.put(DruidDataSourceFactory.PROP_USERNAME, ds.getUsername());
                map.put(DruidDataSourceFactory.PROP_PASSWORD, ds.getPassword());
                if (StringUtils.isNotBlank(ds.getConfig())) {
                    String[] s = ds.getConfig().split("\n");
                    for (String ss : s) {
                        String[] ag = ss.split("\\|");
                        map.put(ag[0], ag[1]);
                    }
                }
                DataSource dataSource = DruidDataSourceFactory.createDataSource(map);
                SpringContextUtil.registerBean(ds.getJdbctemplateName(), JdbcTemplate.class, dataSource, true);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public int load() {
        String sql = "";
        if ("MySQL".equals(wnoConfig.getIntfDB())) {
            sql = "SELECT SOURCE_ID, TYPE, JDBCTEMPLATE_NAME, DRIVER_CLASS_NAME, URL, USERNAME, PASSWORD, CONFIG, STATUS FROM " + wnoConfig.getTablePrefix() + "DBSOURCE WHERE  CREATE_TIME >=  ADDDATE(NOW(), INTERVAL -30 SECOND) ";
        }
        if ("PostgreSQL".equals(wnoConfig.getIntfDB())) {
            sql = "SELECT SOURCE_ID, TYPE, JDBCTEMPLATE_NAME, DRIVER_CLASS_NAME, URL, USERNAME, PASSWORD, CONFIG, STATUS FROM " + wnoConfig.getTablePrefix() + "DBSOURCE WHERE  CREATE_TIME >=  NOW()-interval '30 SECOND' ";
        }
        if ("Oracle".equals(wnoConfig.getIntfDB())) {
            sql = "SELECT SOURCE_ID, TYPE, JDBCTEMPLATE_NAME, DRIVER_CLASS_NAME, URL, USERNAME, PASSWORD, CONFIG, STATUS FROM " + wnoConfig.getTablePrefix() + "DBSOURCE WHERE  CREATE_TIME >=  SYSDATE - 30/(24*60*60) ";
        }
        List<DbSource> dbSourceList = this.jdbcTemplate.query(sql, new BeanPropertyRowMapper<DbSource>(DbSource.class));
        for (DbSource ds : dbSourceList) {
            loadDbSourceBean(ds);
        }
        return dbSourceList.size();
    }

    @Override
    public void updateDbSource(DbSource dbSource) {
        String upSql = "";
        if ("MySQL".equals(wnoConfig.getIntfDB()) || "PostgreSQL".equals(wnoConfig.getIntfDB())) {
            upSql = "UPDATE " + wnoConfig.getTablePrefix() + "DBSOURCE  SET READ_TIME = NOW() WHERE SOURCE_ID = ? ";
        }
        if ("Oracle".equals(wnoConfig.getIntfDB())) {
            upSql = "UPDATE " + wnoConfig.getTablePrefix() + "DBSOURCE  SET READ_TIME = SYSDATE WHERE SOURCE_ID = ? ";
        }
        this.jdbcTemplate.update(upSql, new Object[]{dbSource.getSourceId()});
    }
}
